/**
 *
 *  Copyright (c) 2011, AstroDAbis
 *  All rights reserved.
 *
 *  Redistribution and use in source and binary forms, with or without modification,
 *  are permitted provided that the following conditions are met:
 *
 *      * Redistributions of source code must retain the above copyright notice,
 *        this list of conditions and the following disclaimer.
 *      * Redistributions in binary form must reproduce the above copyright notice,
 *        this list of conditions and the following disclaimer in the documentation
 *        and/or other materials provided with the distribution.
 *
 *  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
 *  ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 *  WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
 *  IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
 *  INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
 *  BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 *  DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 *  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
 *  OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
 *  OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 */
package org.jiscinvolve.astrodabis.galium.ogsadai.activity.server;

import java.util.List;
import java.util.ArrayList;

import uk.org.ogsadai.activity.ActivityProcessingException;
import uk.org.ogsadai.activity.ActivityTerminatedException;
import uk.org.ogsadai.activity.ActivityUserException;
import uk.org.ogsadai.activity.MatchedIterativeActivity;
import uk.org.ogsadai.activity.io.ActivityInput;
import uk.org.ogsadai.activity.io.ActivityPipeProcessingException;
import uk.org.ogsadai.activity.io.BlockWriter;
import uk.org.ogsadai.activity.io.ControlBlock;
import uk.org.ogsadai.activity.io.PipeClosedException;
import uk.org.ogsadai.activity.io.PipeIOException;
import uk.org.ogsadai.activity.io.PipeTerminatedException;
import uk.org.ogsadai.activity.io.TupleListActivityInput;
import uk.org.ogsadai.activity.io.TupleListIterator;
import uk.org.ogsadai.activity.io.TypedActivityInput;
import uk.org.ogsadai.metadata.MetadataWrapper;
import uk.org.ogsadai.tuple.ColumnNotFoundException;
import uk.org.ogsadai.tuple.SimpleTuple;
import uk.org.ogsadai.tuple.Tuple;
import uk.org.ogsadai.tuple.TupleMetadata;

// Added
import uk.org.ogsadai.common.msgs.DAILogger;

import uk.org.ogsadai.tuple.TupleTypes;
import uk.org.ogsadai.tuple.SimpleTupleMetadata;
import uk.org.ogsadai.tuple.ColumnMetadata;
import uk.org.ogsadai.tuple.SimpleColumnMetadata;

import uk.org.ogsadai.metadata.MetadataWrapper;

/**
 * An activity that takes a list of tuples and processes them as an AstroDAbis result set.
 *
 * Based on the example
 * <a href='http://ogsa-dai.sourceforge.net/documentation/ogsadai4.1/ogsadai4.1-axis/ActivityExampleTupleProcessing.html'>TupleProcessingActivity</a> 
 * from the OGSA-DAI documentation.
 * 
 */
public class ResultProcessingActivity
extends MatchedIterativeActivity
    {

    /**
     * Our debug logger.
     * 
     */
    private static final DAILogger log = 
        DAILogger.getLogger(ResultProcessingActivity.class);

    /**
     * Block writer used to write the activity's result output.
     */
    private BlockWriter mResultBlockWriter;

    /**
     * Provides the input details.
     *
     */
    protected ActivityInput[] getIterationInputs()
        {
        return new ActivityInput[] {
            new TypedActivityInput(
                "query",
                String.class
                ),
            new TupleListActivityInput(
                "tuples"
                )
            };
        }
 
    /**
     * Pre-processing.
     *
     */
    protected void preprocess()
        throws ActivityUserException,
               ActivityProcessingException,
               ActivityTerminatedException
        {
        validateOutput("result");
        mResultBlockWriter = getOutput("result");
        }

    /**
     * Process an iteration.
     *
     */
    protected void processIteration(Object[] iterationData)
        throws ActivityProcessingException, 
               ActivityTerminatedException,
               ActivityUserException
        {

        try {

            //
            // Get the query identifier.
            String astroQueryId  = (String) iterationData[0];
            //
            // Get the list of tuples.
            TupleListIterator tupleListIterator = (TupleListIterator) iterationData[1];
            //
            // Get the original metadata.
            MetadataWrapper originalWrapper  = tupleListIterator.getMetadataWrapper();
            TupleMetadata   originalMetadata = (TupleMetadata) originalWrapper.getMetadata();

            //
            // Create the metadata for our new columns.
/*
 * Parameters:
 *  String name - the column name. The name of each column should be unique.
 *  int type - the column type
 *  int precision - the data precision
 *  int nullable - the data nullability. One of ColumnMetadata.COLUMN_NO_NULLS, ColumnMetadata.COLUMN_NULLABLE or ColumnMetadata.COLUMN_NULLABLE_UNKNOWN.
 *  int displaySize - the maximum character width of the column
 *
 */

            List<ColumnMetadata> astroColumns = new ArrayList<ColumnMetadata>();
            astroColumns.add(
                new SimpleColumnMetadata(
                    "query",
                    TupleTypes._STRING,
                    0,
                    ColumnMetadata.COLUMN_NO_NULLS,
                    0
                    )
                );
            astroColumns.add(
                new SimpleColumnMetadata(
                    "row",
                    TupleTypes._LONG,
                    0,
                    ColumnMetadata.COLUMN_NO_NULLS,
                    0
                    )
                );

            //
            // Combine our metadata with the original.
            TupleMetadata astroMetadata = new SimpleTupleMetadata(
                astroColumns
                );
            TupleMetadata combinedMetadata = new SimpleTupleMetadata(
                "AstroDAbis result set",
                astroMetadata,
                originalMetadata
                );
            MetadataWrapper combinedWrapper = new MetadataWrapper(
                combinedMetadata
                );
            //
            // Log the column metadata.
            /*
             */
            for (int colNum = 0 ; colNum < combinedMetadata.getColumnCount() ; colNum++)
                {
                ColumnMetadata colMeta = combinedMetadata.getColumnMetadata(colNum);
                log.debug("ColumnMetadata");
                log.debug("Name  [" + colMeta.getName() + "] ");
                log.debug("Type  [" + colMeta.getType() + "] ");
                log.debug("Table [" + colMeta.getTableName() + "] ");
                }

            //
            // Write the LIST_BEGIN marker and metadata
            mResultBlockWriter.write(ControlBlock.LIST_BEGIN);
            mResultBlockWriter.write(combinedWrapper);
            
            //
            // Process the tuples.
            Tuple tuple ;
            for(long tupNum = 0 ; ((tuple = (Tuple) tupleListIterator.nextValue()) != null) ; tupNum++)
                {
                //
                // Create a new list of elements.
                ArrayList elements = new ArrayList(
                    combinedMetadata.getColumnCount()
                    );
                //
                // Add our extra values.
                elements.add(
                    astroQueryId
                    );
                elements.add(
                    new Long(
                        tupNum
                        )
                    );
                //
                // Add the rest of the data from the original columns.
                int colCount = originalMetadata.getColumnCount();
                for (int colNum = 0; colNum < colCount ; colNum++)
                    {
                    elements.add(
                        tuple.getObject(
                            colNum
                            )
                        );
                    }
                //
                // Write the new tuple to our output.
                mResultBlockWriter.write(
                    new SimpleTuple(
                        elements
                        )
                    );
                }

            //
            // Write the list end marker
            mResultBlockWriter.write(ControlBlock.LIST_END);

            }

        catch (PipeClosedException e)
            {
            // Consumer does not want any more data, just stop.
            }

        catch (PipeIOException e)
            {
            throw new ActivityPipeProcessingException(e);
            }

        catch (PipeTerminatedException e)
            {
            throw new ActivityTerminatedException();
            }
        }

    /**
     * Post-processing.
     *
     */
    protected void postprocess()
        throws ActivityUserException,
               ActivityProcessingException,
               ActivityTerminatedException
        {
        // No post-processing
        }
    }

